package com.g2room.queue.kafka;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
public class KafkaProducerThread extends Thread {

	public static String topic = "gamer_test";
	public static int i = 0;
	public KafkaProducerThread() {
		super();
	}

	@Override
	public void run() {
		Producer<Integer, String> producer = createProducer();
		Long start = System.currentTimeMillis();
		int i = 0;
		while (true) {
			producer.send(new KeyedMessage<Integer, String>(topic, "message: " + i++));
			System.out.println(i);
			int count = 100;
			if(i % count == 0) {
				Long end = System.currentTimeMillis();
				System.out.println(Thread.currentThread().getId() + "，当前数据量=" + i + "，耗时：" + (end -start) + "，每秒数据：" + (count * 1000 / Math.max((end - start), 1)));
				start = end;
			}
			try {
				TimeUnit.SECONDS.sleep(1);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private Producer<Integer, String> createProducer() {
		return new Producer<Integer, String>(new ProducerConfig(getProperties()));
	}
	
	public static Properties getProperties() {
		Properties properties = new Properties();
        // 配置ZK连接
		properties.put("zookeeper.connect", "10.3.246.63:2180,10.3.246.60:2180");
		 //zk连接超时
		properties.put("zookeeper.session.timeout.ms", "4000");
		properties.put("zookeeper.sync.time.ms", "200");
		// 配置序列化类
		properties.put("serializer.class", StringEncoder.class.getName());
//        //配置value的序列化类
//        properties.put("serializer.class", "kafka.serializer.StringEncoder");
        //此处配置的是kafka的集群
		properties.put("metadata.broker.list", "10.3.246.63:9092,10.3.246.60:9092");
//		0，意味着producer永远不会等待一个来自broker的ack，这就是0.7版本的行为。这个选项提供了最低的延迟，但是持久化的保证是最弱的，当server挂掉的时候会丢失一些数据。
//		1，意味着在leader replica已经接收到数据后，producer会得到一个ack。这个选项提供了更好的持久性，因为在server确认请求成功处理后，client才会返回。如果刚写到leader上，还没来得及复制leader就挂了，那么消息才可能会丢失。
//		-1，意味着在所有的ISR都接收到数据后，producer才得到一个ack。这个选项提供了最好的持久性，只要还有一个replica存活，那么数据就不会丢失。
		properties.put("request.required.acks","1");
		return properties;
	}
	
	public static void runKp() {
		KafkaProducerThread kp = new KafkaProducerThread();
		Thread t = new Thread(kp);
		t.start();
	}

	public static void main(String[] args) {
		for(int i=0; i< 1; i ++) {
			runKp();
		}
	}

}
